package com.xzy.rabbit.demo.workqueue.e;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.MessageProperties;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 工作模式（Work Queue）
 *
 * @author xzy
 * @date 2021年9月23日
 */
@Slf4j
@AllArgsConstructor
class Producer implements Runnable {

    /**
     * 开启队列持久化
     */
    private static final boolean DURABLE = true;

    /**
     * 开启消息持久化
     */
    private static final AMQP.BasicProperties PERSISTENT_BASIC = MessageProperties.PERSISTENT_TEXT_PLAIN;

    /**
     * TCP连接
     */
    private final Connection connection;
    /**
     * 队列名称
     */
    private final String queueName;
    /**
     * 消息总数
     */
    private final Integer messageCount;
    /**
     * 延迟时间
     */
    private final long workTime;

    /**
     * 生产消息——队列持久化 + 消息持久化
     */
    public void produce() throws IOException, InterruptedException, TimeoutException {
        // 1.获取信道
        try (Channel channel = connection.createChannel()) {

            // 2.声明队列 + 队列持久化
            channel.queueDeclare(queueName, DURABLE, false, false, null);

            /*
             * 注意：对于一个已经存在且不进行持久化的队列，上述语句会报错。此时可以手动删除队列，然后重新执行代码。
             */

            // 3.发送消息（生产）+ 消息持久化
            String currentThread = Thread.currentThread().getName();
            for (int i = 0; i < messageCount; i++) {
                String message = "msg" + i;

                channel.basicPublish("", queueName, PERSISTENT_BASIC, message.getBytes());
                log.info(currentThread + " 成功发送：{}\n", message);

                Thread.sleep(workTime);
            }
        }
    }
    
    @Override
    public void run() {
        try {
            produce();
        } catch (IOException | InterruptedException | TimeoutException e) {
            e.printStackTrace();
            Thread.currentThread().interrupt();
        }
    }
}